package com.itheima.sh.bulk.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.itheima.sh.bulk.pojo.Sku;
import com.itheima.sh.bulk.mapper.SkuMapper;
import com.itheima.sh.bulk.service.SkuService;
import org.apache.http.HttpHost;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * <p>
 * 商品表 服务实现类
 * </p>
 *
 * @author hl
 * @since 2020-12-29
 */
@Service
public class SkuServiceImpl extends ServiceImpl<SkuMapper, Sku> implements SkuService {

    @Autowired
    private SkuMapper skuMapper;

    @Override
    public Long createIndexAndBulkInsertData() throws IOException, InterruptedException {
        RestHighLevelClient client = new RestHighLevelClient(RestClient.builder(new HttpHost("192.168.200.150", 9200, "http")));
        CreateIndexRequest createIndexRequest = new CreateIndexRequest("sku");
        createIndexRequest.mapping("{\n" +
                "    \"properties\": {\n" +
                "      \"id\":{\n" +
                "        \"type\": \"keyword\"\n" +
                "        , \"store\": true\n" +
                "        , \"index\": false\n" +
                "      },\n" +
                "      \"name\":{\n" +
                "        \"type\": \"text\",\n" +
                "        \"analyzer\": \"ik_max_word\"\n" +
                "        , \"store\": true\n" +
                "        , \"index\": true\n" +
                "      },\n" +
                "      \"price\":{\n" +
                "        \"type\": \"integer\"\n" +
                "        , \"store\": true\n" +
                "        , \"index\": true\n" +
                "      },\n" +
                "      \"num\":{\n" +
                "        \"type\": \"integer\"\n" +
                "        , \"store\": true\n" +
                "        , \"index\": false\n" +
                "      },\n" +
                "      \"images\":{\n" +
                "        \"type\": \"keyword\"\n" +
                "        , \"store\": true\n" +
                "        , \"index\": false\n" +
                "      },\n" +
                "      \"create_time\":{\n" +
                "        \"type\": \"date\"\n" +
                "        , \"store\": true\n" +
                "        , \"index\": true\n" +
                "      },\n" +
                "      \"category_id\":{\n" +
                "        \"type\": \"keyword\"\n" +
                "        , \"store\": true\n" +
                "        , \"index\": false\n" +
                "      },\n" +
                "      \"category_name\":{\n" +
                "        \"type\": \"text\"\n" +
                "        ,\"analyzer\": \"ik_max_word\"\n" +
                "        , \"store\": true\n" +
                "        , \"index\": true\n" +
                "      },\n" +
                "      \"brand_name\":{\n" +
                "        \"type\": \"keyword\"\n" +
                "        , \"store\": true\n" +
                "        , \"index\": true\n" +
                "      }\n" +
                "    }\n" +
                "  }", XContentType.JSON);
        Map<String, Object> map = new HashMap<>();
        map.put("number_of_shards", 3);
        map.put("number_of_replicas", 1);
        createIndexRequest.settings(map);
        client.indices().create(createIndexRequest, RequestOptions.DEFAULT);
        long start = System.currentTimeMillis();
        Integer number = 5000;
        AtomicInteger atomicInteger = new AtomicInteger(0);
        for (int i = 0; i < 19; i++) {
            Integer startIndex = i * 5000;
            new Thread(new Runnable() {
                @Override
                public void run() {
                    List<Sku> skuList = skuMapper.findDataLimit(startIndex, number);
                    BulkRequest bulkRequest = new BulkRequest("sku");
                    for (Sku sku : skuList) {
                        IndexRequest indexRequest = new IndexRequest();
                        indexRequest.id(sku.getId());
                        String skuJson = JSON.toJSONString(sku);
                        indexRequest.source(skuJson, XContentType.JSON);
                        bulkRequest.add(indexRequest);
                    }
                    try {
                        client.bulk(bulkRequest, RequestOptions.DEFAULT);
                        atomicInteger.getAndSet(atomicInteger.get() + 1);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        }
        long time = 0L;
        while (atomicInteger.get() != 19) {
           Thread.sleep(10);
        }
        long end = System.currentTimeMillis();
        time = end - start;
        return time;
    }
}
